JDBC Databases

Relational Database Management Systems (RDBMS) like PostgreSQL, MySQL, and Oracle are major sources of enterprise transactions. Spark can connect directly to these databases using Java Database Connectivity (JDBC) drivers, enabling parallelized, distributed data reads and writes.

The Danger of Default JDBC Reads

By default, when you read a database table via: spark.read.format("jdbc").option("dbtable", "users").load()

• Spark runs a single task (using only a single executor thread on a single machine) to fetch the entire database table.
• This creates a severe bottleneck on your Spark cluster and can swamp your RDBMS database with a single massive query.

Parallelizing JDBC Ingestions (Best Practice)

To read database tables in parallel using multiple executors, you must instruct Spark how to partition the query by specifying:

1. partitionColumn: A numeric or date column (e.g. user_id or created_at).
2. lowerBound & upperBound: The minimum and maximum boundaries of that column.
3. numPartitions: The target number of concurrent database connection threads.

Spark will automatically divide the table into numeric chunks and execute parallel queries: SELECT * FROM users WHERE id >= 0 AND id < 10000, SELECT * FROM users WHERE id >= 10000...

PySpark Code Example: Parallel Reading & Batch Writing

Here is a complete template showcasing how to read from a relational database in parallel and write back with custom batch sizes:

from pyspark.sql import SparkSession

# 1. Setup Spark including the database JDBC driver jar dependency
# Note: Ensure you include the driver coordinates (e.g. postgresql driver) in your submit script
spark = SparkSession.builder \
    .appName("JDBC Databases") \
    .config("spark.jars.packages", "org.postgresql:postgresql:42.6.0") \
    .master("local[*]") \
    .getOrCreate()

# 2. Database Connection Options
db_properties = {
    "url": "jdbc:postgresql://localhost:5432/enterprise_warehouse",
    "user": "db_user",
    "password": "db_secure_password",
    "driver": "org.postgresql.Driver"
}

# 3. Parallel Ingestion of a massive table
# We divide the ingestion into 4 parallel connection threads based on employee id boundaries
df = spark.read \
    .format("jdbc") \
    .option("url", db_properties["url"]) \
    .option("user", db_properties["user"]) \
    .option("password", db_properties["password"]) \
    .option("driver", db_properties["driver"]) \
    .option("dbtable", "employees") \
    .option("partitionColumn", "employee_id") \
    .option("lowerBound", "1") \
    .option("upperBound", "100000") \
    .option("numPartitions", "4") \
    .load()

df.show()

# 4. Write data back to a database table in batches
# We configure batchsize and isolation levels to optimize transaction speeds
df.write \
    .format("jdbc") \
    .option("url", db_properties["url"]) \
    .option("user", db_properties["user"]) \
    .option("password", db_properties["password"]) \
    .option("driver", db_properties["driver"]) \
    .option("dbtable", "employees_backup") \
    .option("batchsize", "5000") \
    .option("isolationLevel", "NONE") \
    .mode("overwrite") \
    .save()